package org.apache.rocketmq.example.quickstart.transaction;

import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.*;

/**
 * 事务生产者
 */
public class TransactionProducer {
    public static void main(String[] args) throws Exception {
        // 创建生产者回查线程池
        ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueue<>(2), new ThreadFactory() {
            @Override
            public Thread newThread(Runnable runnable) {
                Thread thread = new Thread(runnable);
                thread.setName("client-transaction-msg-check-thread");
                return thread;
            }
        });

        // 创建事务消息生产者
        TransactionMQProducer producer = new TransactionMQProducer("producer-group-1");
        producer.setNamesrvAddr("39.107.66.31:9876");
        producer.setExecutorService(executorService); // 设置生产者回查线程池
        producer.setTransactionListener(new TransactionListenerImpl()); // 设置生产者监听器
        producer.start();

        // 发送半事务
        try {
            Message message = new Message("topic-10", null, "事务数据".getBytes());
            TransactionSendResult sendResult = producer.sendMessageInTransaction(message, null);
            // 解析发送结果
            SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            System.out.println(sendResult.getSendStatus() + "-" + format.format(new Date()));
        } catch (Exception e) {
            // 发送失败，回滚事务
        } finally {
//            producer.shutdown();
        }
    }
}
